package com.smsc.headend.connector.service.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.lang.Tuple;
import cn.hutool.core.util.HexUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.google.common.base.Preconditions;
import com.google.common.primitives.Bytes;
import com.smsc.headend.common.enums.DeviceType;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.common.utils.RedisKeys;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.connector.annotations.DeviceSignInLimiter;
import com.smsc.headend.connector.bo.*;
import com.smsc.headend.connector.builder.LoggerBuilder;
import com.smsc.headend.connector.enums.consts.NettyConstant;
import com.smsc.headend.connector.enums.states.ChannelState;
import com.smsc.headend.connector.exception.UDISDeviceConException;
import com.smsc.headend.connector.manager.AssociationManager;
import com.smsc.headend.connector.manager.DeviceKeyManager;
import com.smsc.headend.connector.manager.TCPClientManager;
import com.smsc.headend.connector.manager.TaskTicketManager;
import com.smsc.headend.connector.netty.NettyConfigHelper;
import com.smsc.headend.connector.netty.SyncFuture;
import com.smsc.headend.connector.service.*;
import com.smsc.headend.connector.service.asset.AssetService;
import com.smsc.headend.connector.service.protocol.AnsiPacketCodecService;
import com.smsc.headend.connector.service.protocol.DLMSPacketCodecService;
import com.smsc.headend.connector.utils.CosemXMLUtil;
import com.smsc.headend.connector.utils.LogFileNameUtil;
import com.smsc.headend.connector.utils.NetUtil;
import com.smsc.headend.connector.utils.PacketAddressUtil;
import com.smsc.headend.connector.utils.concurrent.TaskCountDownLatch;
import com.smsc.headend.module.asset.dmo.ComConfigDMO;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.event.dto.DeviceIpChangeEventDto;
import com.smsc.headend.module.mdm.dds.dto.MdmAnonyTerminalDto;
import com.smsc.headend.module.mdm.dds.enums.MdmSourceEnum;
import com.smsc.headend.module.protocol.ProtocolType;
import com.smsc.headend.module.protocol.ansi.AnsiTranslator;
import com.smsc.headend.module.protocol.ansi.enums.AnsiErrorCode;
import com.smsc.headend.module.protocol.ansi.objects.Acse;
import com.smsc.headend.module.protocol.ansi.utils.AnsiBaseUtil;
import com.smsc.headend.module.protocol.cosem.util.ByteNumberUtils;
import com.smsc.headend.module.protocol.cosem.util.ByteUtils;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.ConnectorDeviceNotification;
import com.smsc.headend.module.task.dto.ConnectorTask;
import com.smsc.headend.module.task.dto.ConnectorTaskResult;
import com.smsc.headend.module.task.proto.ConnectorDeviceNotificationProto;
import com.smsc.headend.module.task.proto.ConnectorTaskResultProto;
import gurux.dlms.GXByteBuffer;
import gurux.dlms.GXDLMSClient;
import gurux.dlms.GXDLMSTranslator;
import gurux.dlms.TranslatorOutputType;
import gurux.dlms.enums.InterfaceType;
import gurux.dlms.secure.GXDLMSSecureClient;
import io.netty.channel.Channel;
import io.netty.util.Attribute;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class DeviceConnectServiceImpl implements DeviceConnectService {
    public static final int HEART_BEAT_VALID_SECONDS = 21600;//6 hours
    public static final String CHANNEL = "channel";
    public static final String LOCK_COM_ENTITY_FAILED = "lockComEntityFailed";

    @Autowired
    RedisUtils redisUtils;
    @Autowired
    FeignAssetManagementService assetManagementRemoteService;
    @Autowired
    TaskEngineService taskEngineService;
    @Autowired
    DLMSPacketCodecService dlmsPacketCodecService;
    @Autowired
    AnsiPacketCodecService ansiPacketCodecService;
    @Autowired
    TCPClientManager tcpClientManager;
    @Autowired
    TerminalChannelService terminalChannelService;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    AnonyDeviceService anonyDeviceService;
    @Autowired
    LogTerminalAccessService logTerminalAccessService;
    @Autowired
    @Lazy
    DeviceConnectService selfService;

    @Autowired
    TaskTicketManager taskTicketManager;

    @Autowired
    DeviceStatusService deviceStatusService;

    @Value("${ports.dcu_port:6603}")
    String dcuPort;

    @Value("${ports.p2p_port:6604}")
    String p2pPort;

    @Autowired
    AssociationManager associationManager;

    @Autowired
    TaskService taskService;

    @Autowired
    LogFileNameUtil logFileNameUtil;

    @Autowired
    DeviceConnectService deviceConnectService;

    @Autowired
    DeviceKeyManager deviceKeyManager;

    @Autowired
    AssetService assetService;

    GXDLMSClient shareClient;

    ThreadLocal<ConnectionInfo> connectionInfo = new ThreadLocal<ConnectionInfo>();

    private Map<Long, TaskCountDownLatch> comLatchMap = new ConcurrentHashMap<>();

    private static final String COM_ID = "comId";

    public ThreadLocal<ConnectionInfo> getConnectionInfoThreadLocal() {
        return connectionInfo;
    }

    @PostConstruct
    public void initShareClient() {
        shareClient = new GXDLMSClient();
        shareClient.setInterfaceType(InterfaceType.WRAPPER);
    }

    @Override
    public void deviceDisconnect(Channel channel) {
        ComEntity comEntity = channel.attr(NettyConfigHelper.comEntityAttrKey).get();
        ConnectorTask connectorTask = channel.attr(NettyConfigHelper.connectTaskRegular).get();
        if (connectorTask != null && connectorTask.getTicketNumber() != null) {
            taskTicketManager.recycleTicket(connectorTask.getTicketNumber());
        }
        if (comEntity != null && comEntity.getComId() != -1L) {
//            logTerminalAccessService.addAccessLog(comEntity.getComId(), LogComAccess.AccessType.Disconnect.getType());
            deviceStatusService.refreshDeviceStatus(comEntity, null, ComEntity.CommunicationStatus.OFFLINE);
            assetManagementRemoteService.updateComEntityStatus(comEntity.getComId(), 2, SystemClock.now() / org.apache.commons.lang3.time.DateUtils.MILLIS_PER_SECOND);
//            comEntityStatusManager.comEntityOffline(comEntity, SystemClock.now() / 1000);
        }
    }

    @Override
    public void sendToDevice(ConnectorTask connectorTask) {
        MDC.put(COM_ID, String.valueOf(connectorTask.getTerminalId()));
        MDC.put("meterId", String.valueOf(connectorTask.getMeterId()));
        MDC.put("atomicTaskNo", connectorTask.getAtomicTaskNo());
//        log.info("{}, terminal {} meter {}", connectorTask.getAtomicTaskNo(), connectorTask.getTerminalId(), connectorTask.getMeterId());
        Long terminalId = connectorTask.getTerminalId();
        ComEntity terminal = terminalChannelService.getTerminalById(terminalId);
        redisUtils.set(RedisKeys.getTaskLastAliveTime(connectorTask.getTaskNo()), Convert.toStr(SystemClock.now()), 600);
        if (terminal == null || terminal.getComId() == null) {
            log.debug("terminal {} not exists ", terminalId);
            return;
        }
        try {
            log.debug("TryAcquireLatch: comId:{}", connectorTask.getTerminalId());
            acquireComEntityLatch(connectorTask.getTerminalId(), connectorTask.getTaskNo());
        } catch (UDISDeviceConException e) {
            this.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, e.getErrorText(), terminal.getComId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), terminal);
        } catch (Exception e) {
            LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(terminal), this.getClass()).error("unexpected error", e);
            this.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "unknowError", terminal.getComId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), terminal);
        }
        if (connectorTask.getRoleId() == null || connectorTask.getRoleId() == 0) {
            log.error("task with no roleId: {}", JSONUtil.toJsonStr(connectorTask));
            this.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "data item no role id", terminal.getComId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), terminal);
            return;
        }
        connectionInfo.set(new ConnectionInfo(connectorTask.getTaskNo(), connectorTask.getTaskNo(), terminal.getModuleSn(), terminal.getComId(), connectorTask.getMeterId()));
        Channel channel = NettyConfigHelper.getChannelByComId(terminalId);
        String netAddress = NetUtil.getIpFromChannel(channel);
        log.debug("comId ={} getChannel={}", terminal.getComId(), netAddress);
        Boolean isSendPacketSync = false;
        //客户端模式
        if (ComEntity.CommunicationMode.SERVER.getCode() == terminal.getComMode()) {
            if (channel == null || !channel.isActive()) {
//                LoggerBuilder.getLogger(terminal.getModuleSn(), this.getClass()).error("{} tcp channel not exists", terminalId);
//                sendDeviceResultMessage(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "tcp channel not exists");
                return;
            }
            isSendPacketSync = true;
        }
        //服务端模式
        else {
            if (channel != null && channel.isActive()) {
                isSendPacketSync = true;
            }
            if (channel == null || !channel.isActive()) {
                connectToServerDevice(terminal, connectorTask);
                isSendPacketSync = false;
            }
        }

        if (isSendPacketSync && channel != null) {
            MDC.put(CHANNEL, NetUtil.getIpFromChannel(channel));
            sendByProtocol(channel.attr(NettyConfigHelper.protocol).get(), connectorTask, channel);
        }
    }

    private Boolean acquireComEntityLatch(Long terminalId, String taskNo) throws UDISDeviceConException {
        TaskCountDownLatch comEntityCountDown = getComEntityResourceLatch(terminalId, taskNo);
        if (comEntityCountDown.getCount() == 1 && !comEntityCountDown.getTaskNo().equalsIgnoreCase(taskNo)) {
            try {
                boolean waitSuccess = comEntityCountDown.await(180, TimeUnit.SECONDS);
                log.debug("WaitLatch:comId:{},result:{}", terminalId, waitSuccess);
                if (!waitSuccess) {
                    log.error("WaitLatchOverTime:comId={},taskNo={}", terminalId, taskNo);
                    throw new UDISDeviceConException(LOCK_COM_ENTITY_FAILED);
                }
            } catch (InterruptedException e) {
                log.error("ComEntityResourceGetFailed:comId={},taskNo={}", terminalId, taskNo);
                throw new UDISDeviceConException(LOCK_COM_ENTITY_FAILED);
            }
        }
        Boolean competeResult = acquireComEntityResource(terminalId, taskNo);
        if (competeResult) {
            log.debug("AcquireLockSuccess:{}", taskNo);
            return true;
        }
        while (!competeResult) {
            competeResult = acquireComEntityLatch(terminalId, taskNo);
        }
        return true;
    }

    public synchronized TaskCountDownLatch getComEntityResourceLatch(Long terminalId, String taskNo) {
        if (!comLatchMap.containsKey(terminalId)) {
            log.debug("createLatch,comId:{},taskNo:{}", terminalId, taskNo);
            TaskCountDownLatch countDownLatch = new TaskCountDownLatch(0, taskNo);
            comLatchMap.put(terminalId, countDownLatch);
        }
        TaskCountDownLatch countDownLatch = comLatchMap.get(terminalId);
        return countDownLatch;
    }

    private synchronized Boolean acquireComEntityResource(Long terminalId, String taskNo) {
        TaskCountDownLatch countDownLatch = comLatchMap.get(terminalId);
        log.debug("getLatch:{}", JSONUtil.toJsonStr(countDownLatch));
        if (countDownLatch != null && taskNo.equals(countDownLatch.getTaskNo())) {
            //Task重复发送过来时 重新获取Latch
            if (countDownLatch.getCount() == 0) {
                TaskCountDownLatch cdl = new TaskCountDownLatch(1, taskNo);
                comLatchMap.put(terminalId, cdl);
            }
            return true;
        }
        if (countDownLatch != null && countDownLatch.getCount() == 1) {
            return false;
        }
        log.debug("accLatchAndCreateNew,comId:{},taskNo:{}", terminalId, taskNo);
        TaskCountDownLatch countDownLatch1 = new TaskCountDownLatch(1, taskNo);
        comLatchMap.put(terminalId, countDownLatch1);
        return true;
    }

    public void sendByProtocol(ProtocolType protocol, ConnectorTask connectorTask, Channel channel) {
        if (protocol == null) {
            try {
                protocol = setChannelProtocol(channel);
            } catch (Exception e) {
                selfService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "unknown error"
                        , connectorTask.getTerminalId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), channel.attr(NettyConfigHelper.comEntityAttrKey).get());
                return;
            }
        }

        switch (protocol) {
            case DLMS_Gateway_Protocol:
            case DLMS_COSEM_46_Protocol:
            case DLMS_COSEM_47_Protocol:
                dlmsPacketCodecService.sendPacket(connectorTask, channel);
                break;
            case ANSI_C12_18_Protocol:
                ansiPacketCodecService.sendPacket(connectorTask, channel);
                break;
            default:
                throw new IllegalStateException("not support protocol:" + protocol);
        }
    }

    private ProtocolType setChannelProtocol(Channel channel) {
        ComEntity com = channel.attr(NettyConfigHelper.comEntityAttrKey).get();
        ComConfigDMO comConfigById = assetManagementRemoteService.getComConfigById(com.getComId());
        ProtocolType type = ProtocolType.getType(comConfigById.getProtocolId());
        channel.attr(NettyConfigHelper.protocol).set(type);
        return type;
    }

    private void connectToServerDevice(ComEntity terminal, ConnectorTask connectorTask) {
        Boolean connecting = tcpClientManager.isConnectedInOtherNode(terminal.getComId());
        String getenv = System.getenv("local.server.port");
        if (connecting) {
            log.debug("DeviceServer notIn:{},  terminalId:{}", getenv, terminal.getComId());
            return;
        }
        //get redis lock
        Long lockKey = redisUtils.easyRedisLock(terminal.getComId().toString());
        if (lockKey != null) {
            try {
                String netAddr = terminal.getIpAddress();

                if (StringUtils.isEmpty(netAddr)) {
                    LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(terminal), this.getClass()).error("server type ComEntity have no NetAddr");
                    sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "SERVER_NO_NETADDR", terminal.getComId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), terminal);
                    return;
                }
                ComConfigDMO comConfigById = assetManagementRemoteService.getComConfigById(terminal.getComId());
                tcpClientManager.connectAndSendTask(terminal, comConfigById.getProtocolId(), connectorTask);
            } finally {
                try {
                    redisUtils.unlockEasyRedisLock(terminal.getComId().toString(), lockKey);
                } catch (Exception e) {
                    log.error("release error");
                }
            }
        }
        return;
    }

    @Override
    public void ansiPacketAccept(Channel channel, byte[] result) {
        ChannelState channelState = channel.attr(NettyConfigHelper.channelState).get();
        switch (channelState) {
            case Init:
                handleSignInPacket(channel, result);
                break;
            case Registered:
                handleBusinessPacket(channel, result);
                break;
            default:
                log.error("channelUnknownState:{}", channelState);
        }
    }

    @Override
    public void sendNotificationToUTE(String pduXml, Channel channel, String deviceAddress) {
        ComEntity com = channel.attr(NettyConfigHelper.comEntityAttrKey).get();
        if (com == null) {
            return;
        }
        ConnectorDeviceNotification notification = new ConnectorDeviceNotification();
        notification.setComId(com.getComId());
        notification.setDeviceAddress(deviceAddress);
        notification.setData(pduXml);
        notification.setType(ConnectorDeviceNotification.NotificationType.Event.getType());
        byte[] notificationResponse = dlmsPacketCodecService.getNotificationResponse(pduXml);
        if (notificationResponse != null) {
            sendPacket2Channel(channel, notificationResponse);
        }
        kafkaTemplate.send(TaskKafkaTopic.CON_DEVICE_NOTIFICATION, ProtoBufferUtil.toProtoBuffer(notification, ConnectorDeviceNotificationProto.ConnectorDeviceNotification.class).toByteArray());
    }

    @Override
    public void unlockComEntity(Channel channel) {
        ComEntity com = channel.attr(NettyConfigHelper.comEntityAttrKey).get();
        if (com == null || com.getComId() == -1) {
            return;
        }
        CountDownLatch comEntityResourceLatch = getComEntityResourceLatch(com.getComId(), null);
        if (comEntityResourceLatch != null) {
            comEntityResourceLatch.countDown();
        } else {
            log.error("comIdExecuteTaskWithoutLatch:ComId:{}", com.getComId());
        }
    }

    public void handleSignInPacket(Channel channel, byte[] result) {
        channel.attr(NettyConfigHelper.deviceLgcPkg).setIfAbsent(result);
        byte[] response = selfService.deviceChannelEstablish(channel, result);
        if (response != null && response.length > 0) {
            sendPacket2Channel(channel, response);
        }
    }

    private void handleBusinessPacket(Channel channel, byte[] result) {
        ComEntity com = channel.attr(NettyConfigHelper.comEntityAttrKey).get();
        Boolean isHeartBeat = ansiPacketCodecService.isHeartbeat(result);
        if (isHeartBeat) {
            byte[] hbResponse = ansiPacketCodecService.heartbeatResponse(com, result);
            if (hbResponse != null) {
                selfService.sendPacket2Channel(channel, hbResponse);
            }
            //TODO: ANSI数据少 等性能优化代码合并后，修改成与DLMS一致
            redisUtils.set(RedisKeys.getComLastHeartbeat(com.getComId()), String.valueOf(SystemClock.now() / 1000), HEART_BEAT_VALID_SECONDS);
            return;
        }
        ConnectorTask connectorTask = channel.attr(NettyConfigHelper.connectTaskRegular).get();
        if (connectorTask == null) {
            return;
        }
        Acse acse;
        try {
            acse = AnsiTranslator.pduToBean(result);
            if (acse == null) {
                deviceConnectService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "decode fail", connectorTask.getTerminalId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), com);
                return;
            }
        } catch (Exception e) {
            deviceConnectService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "decode fail:" + e.getMessage(), connectorTask.getTerminalId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), com);
            return;
        }

        // 打印解密后日志
        LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(com), this.getClass()).info("after encrypt: <--{}", ByteUtils.byteToHexString(acse.getPacket()));
        byte flag = acse.getPdu()[0];
        //重试任务结束
        String netAddress = NetUtil.getIpFromChannel(channel);
        resetResendTask(netAddress);
        logTerminalAccessService.savePacket(ansiPacketCodecService.buildConnectInfo(connectorTask), SystemClock.now(), result, Boolean.FALSE);
        //Common Error
        if (flag != 0) {
            AnsiErrorCode code = AnsiErrorCode.getCode(flag);
            selfService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, code.getDesc(), connectorTask.getTerminalId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), com);
            return;
        }
        String s = new String(acse.getPdu(), StandardCharsets.ISO_8859_1);
        Map<String, String> resultMap = new HashMap<>();
        ProtocolType protocolType = channel.attr(NettyConfigHelper.protocol).get();
        resultMap.put(String.valueOf(protocolType.getProtocolId()), s);
        selfService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.success, connectorTask.getAtomicTaskNo(), JSONUtil.toJsonStr(resultMap), "success", connectorTask.getTerminalId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), com);
    }

    @Override
    public void dlmsPacketAccept(Channel channel, byte[] result) {
        ChannelState channelState = channel.attr(NettyConfigHelper.channelState).get();
        switch (channelState) {
            case Init:
                handleSignInPacket(channel, result);
                break;
            case WaitingReg:
                log.info("channelWaitingRegister:{}", NetUtil.getIpFromChannel(channel));
                break;
            case Registered:
                PacketHandleResultBO response = doDlmsPacketAccept(channel, result);
                if (response.getResponsePacket() != null) {
                    selfService.sendPacket2Channel(channel, response.getResponsePacket());
                }
                break;
            default:
                log.error("channelUnknownState:{}, {}", channelState, NetUtil.getIpFromChannel(channel));
        }
    }

    public int lazySignIn(Channel channel) {
        byte[] loginPacket = channel.attr(NettyConfigHelper.deviceLgcPkg).get();
        if (loginPacket == null) {
            try {
                log.info("kickOffChannel:{}", NetUtil.getIpFromChannel(channel));
                channel.attr(NettyConfigHelper.closeBySystem).set(true);
                channel.disconnect().sync();
            } catch (Exception e) {
                log.error("closeException:{}", NetUtil.getIpFromChannel(channel), e);
            }
            return 0;
        }
        byte[] bytes = selfService.deviceChannelEstablish(channel, loginPacket);
        return bytes == null ? 0 : 1;
    }

    public PacketHandleResultBO doDlmsPacketAccept(Channel channel, byte[] cosemApdu) {
        ComEntity comEntity = channel.attr(NettyConfigHelper.comEntityAttrKey).get();
        String netAddress = NetUtil.getIpFromChannel(channel);
        MDC.put(CHANNEL, netAddress);
        PacketHandleResultBO resultBO = new PacketHandleResultBO();
        //加入日志需要的信息
        connectionInfo.set(new ConnectionInfo().setComId(comEntity.getComId()).setModuleSn(comEntity.getModuleSn()));
        MDC.put(COM_ID, String.valueOf(comEntity.getComId()));
        MDC.put("meterId", "");
        if (dlmsPacketCodecService.isHeartBeat(cosemApdu)) {
            return resultBO;
        }
//        LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(comEntity), this.getClass()).info("<--: {}", ByteUtils.byteToHexString(cosemApdu));
        //DLMSGateway返回的透传失败报文做抛弃处理，等待重试
        if (dlmsPacketCodecService.isDLMSGatewayTransparentFailedPkg(cosemApdu)) {
            LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(comEntity), this.getClass()).info("Transparent reading return empty pdu!");
            return resultBO;
        }
        if (dlmsPacketCodecService.isChannelSpecialPkg(cosemApdu)){
            Attribute<ConnectorTask> attribute = NettyConfigHelper.getChannelByAddress(netAddress).attr(NettyConfigHelper.connectTaskKey);
            ConnectorTask task = attribute.getAndSet(null);
            if (task != null && (task.getFlag() & 2) == 2) {
                final ConnectionInfo fiCon = this.connectionInfo.get();
                fiCon.setReturnPacket(task.getReturnPacket()).setTaskNo(task.getTaskNo());
                if (fiCon.getReturnPacket()) {
                    logTerminalAccessService.savePacket(fiCon, SystemClock.now(), cosemApdu, Boolean.FALSE);
                }

                selfService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.success, task.getAtomicTaskNo(), String.format("<Raw Value=\"%s\"/>", ByteUtils.byteToHexString(cosemApdu)), null, comEntity.getComId(), task.getFlag(), task.getTicketNumber()), comEntity);
                return resultBO.setComplete(true).setResponsePacket(cosemApdu);
            }
        }

        //心跳以外的请求设置重试任务结束

        String cosemMessage = dlmsPacketCodecService.setCosemMessage(cosemApdu);
        if (cosemMessage == null) {
            dlmsPacketCodecService.handleUntranslatablePacket(cosemApdu, channel);
            return resultBO;
        }
        String responseType = CosemXMLUtil.getResponseType(cosemMessage);
        ProtocolType protocolType = channel.attr(NettyConfigHelper.protocol).get();

        //处理ReleaseResponse
        if (dlmsPacketCodecService.isAssociationRelease(cosemApdu, netAddress, responseType, protocolType)) {
            resetChannelRetry(channel);
            boolean resent = dlmsPacketCodecService.resendConnectorTask(comEntity, channel);
            if (resent) {
                return resultBO;
            }
        }
        //处理推送报文
        Boolean isNotification = dlmsPacketCodecService.acceptHandleNotifications(responseType, channel, cosemApdu);
        if (isNotification) {
            return resultBO;
        }
        resetChannelRetry(channel);
        //处理业务报文, 非重试任务的AA Release
        resultBO = doHandleDevicePacket(netAddress, comEntity, cosemApdu, channel);
        return resultBO;
    }

    private void resetChannelRetry(Channel channel) {
        channel.attr(NettyConfigHelper.resendPkg).set(null);
        ScheduledFuture future = channel.attr(NettyConfigHelper.retryFuture).getAndSet(null);
        if (future != null && !future.isCancelled()) {
            future.cancel(Boolean.TRUE);
        }
    }

    private void resetResendTask(String netAddress) {
        Channel channel = NettyConfigHelper.getChannelByAddress(netAddress);
        Preconditions.checkNotNull(channel);
        channel.attr(NettyConfigHelper.resendPkg).set(null);
        ScheduledFuture future = channel.attr(NettyConfigHelper.retryFuture).getAndSet(null);
        if (future != null && !future.isCancelled()) {
            future.cancel(Boolean.TRUE);
        }
    }

    /**
     * 由系统下发，未产生
     *
     * @param netAddress
     * @param terminal
     * @param byteReceive
     * @param channel
     * @return
     */
    private PacketHandleResultBO doHandleDevicePacket(String netAddress, ComEntity terminal, byte[] byteReceive, Channel channel) {
        PacketHandleResultBO resultBO = new PacketHandleResultBO();
        resultBO.setLoggerName(logFileNameUtil.getLogFileName(terminal));
        PacketType pt = null;
        //获取channel下电表号， DLMS_Gateway的透传报文，第9位为 0xE7
        String deviceAddress = PacketAddressUtil.getMeterAddress(byteReceive);

//        log.info("get context netaddr={},{}", netAddress, taskContext);
        ConnectorTask connectorTask = channel.attr(NettyConfigHelper.connectTaskRegular).get();
        if (connectorTask == null) {
            log.error("recevieByteUnknown {}", ByteUtils.byteToHexString(byteReceive));
            resultBO.setComplete(true);
            return resultBO;
        }
        //flag & 2 > 0 的任务直接返回RawData
        if ((connectorTask.getFlag() & 2) == 2) {
            final ConnectionInfo fiCon = this.connectionInfo.get();
            fiCon.setReturnPacket(connectorTask.getReturnPacket()).setTaskNo(connectorTask.getTaskNo());
            if (fiCon.getReturnPacket()) {
                logTerminalAccessService.savePacket(fiCon, SystemClock.now(), byteReceive, Boolean.FALSE);
            }
            selfService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.success, connectorTask.getAtomicTaskNo(), String.format("<Raw Value=\"%s\"/>", HexUtil.encodeHex(byteReceive)), null, terminal.getComId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), terminal);
            return resultBO.setComplete(true);
        }

        AssociationState associationState = associationManager.getApplicationAssociationState(netAddress, deviceAddress, connectorTask.getRoleId());
        Boolean aaBuilding = associationManager.isAuthoring(associationState);
        if (associationState != null) {
            associationState.setLastResponseTime(SystemClock.now());
        }

        if (aaBuilding) {
            handleDlmsDeviceAssociationProcess(terminal, associationState, channel, byteReceive);
            return resultBO;
        }
        TaskContext taskContext = terminalChannelService.getTaskContext(netAddress, deviceAddress, terminal.getComId());
        if (taskContext == null) {
            //收到的报文 没有上下文 非心跳， 非注册帧
            pt = pt == null ? getPacketType(byteReceive, terminal) : pt;
            if (!PacketType.Type.DataNotification.equals(pt.getType()) && !PacketType.Type.EventNotificationRequest.equals(pt.getType())) {
                LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(terminal), this.getClass()).error("get packet with no context: packet={}", ByteUtils.byteToHexString(byteReceive));
                return resultBO.setComplete(true);//清除缓存的报文
            }
        }

        String atomicTaskNo = taskContext != null ? taskContext.getAtomicTaskNo() : null;
        Long meterId = taskContext != null ? taskContext.getMeterId() : null;
        MDC.put("atomicTaskNo", atomicTaskNo);
        MDC.put("meterId", String.valueOf(meterId));
        LoggerBuilder.getLogger(logFileNameUtil.getConnectInfoLogFileName(this.connectionInfo.get()), this.getClass()).debug("atomicTask {}, context={}", atomicTaskNo, taskContext);

        //设置connectionInfo信息
        if (taskContext != null) {
            connectionInfo.get().setReturnPacket(taskContext.getReturnPacket()).setAtomicTaskNo(atomicTaskNo).setTaskNo(taskContext.getTaskNo()).setMeterId(meterId);
        }

//        Long now = SystemClock.now();
//        //记录AA最后一次收到response的时间
//        AssociationState associationState = associationManager.getApplicationAssociationState(netAddress, taskContext.getDeviceAddress(), null);
//        if (associationState == null) {
//            LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(terminal), this.getClass()).error("receive packet with no aa state");
//        } else {
//            associationState.setLastResponseTime(SystemClock.now());
//        }

        final ConnectionInfo fiCon = connectionInfo.get();
        //sync request response
        UnwrappedPacket unwrappedPacket = dlmsPacketCodecService.unwrapPrefixPacket(byteReceive);
        if (unwrappedPacket == null) {
            LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(terminal), this.getClass()).error("unwrap packet fail: content={}", ByteUtils.byteToHexString(byteReceive));
            return resultBO.setComplete(true);
        }
        byte[] newByteReceive = Arrays.copyOf(unwrappedPacket.getPacket(), unwrappedPacket.getPacket().length);

        try {
            if (taskContext.getClient() instanceof GXDLMSSecureClient && dlmsPacketCodecService.needEncrypt((GXDLMSSecureClient) taskContext.getClient())) {
                newByteReceive = dlmsPacketCodecService.decryptHLSPacket(newByteReceive, (GXDLMSSecureClient) taskContext.getClient());
            }
        } catch (Exception e) {
            log.error("decrypt packet error", e);
        }

        if (fiCon.getReturnPacket()) {
            byte[] finalBytes = newByteReceive;
            logTerminalAccessService.savePacket(fiCon, SystemClock.now(), finalBytes, Boolean.FALSE);
        }
        DevicePacketHandleResult result = dlmsPacketCodecService.handleDeviceServicePacket(channel, terminal, netAddress, byteReceive, taskContext);
        if (result.getConnectTaskResultDto() != null && result.getConnectTaskResultDto().getExecuteCode().getCode() != ConnectorTaskResult.ExecuteCode.ExceptionResponse.getCode()) {
            selfService.sendDeviceResultMessage(result.getConnectTaskResultDto(), terminal);
        }
        if (result.getResponse() != null) {
            boolean isHdlc = associationState.getClient().getInterfaceType().equals(InterfaceType.HDLC);
            dlmsPacketCodecService.configRetryBO(isHdlc, channel, result.getResponse(), connectorTask, deviceAddress);
        }
        return resultBO.setResponsePacket(result.getResponse()).setComplete(true);
    }

    private void handleDlmsDeviceAssociationProcess(ComEntity terminal, AssociationState associationState, Channel channel, byte[] byteReceive) {

        String cosemXml = dlmsPacketCodecService.getCosemXml();
        if (StringUtils.isEmpty(cosemXml)) {
            return;
        }
        if ("ReleaseResponse".equalsIgnoreCase(CosemXMLUtil.getResponseType(cosemXml))) {
            return;
        }
        Integer aaAccepted = associationManager.packetReceive(channel, terminal, associationState, byteReceive);
        if (aaAccepted == 0) {
            return;
        }
        if (aaAccepted == 1) {
            byte[] businessPacket = dlmsPacketCodecService.initialTaskAndSend(terminal, associationState, channel, Boolean.TRUE);
            associationState.setState(AssociationState.State.authorized);
            boolean isHdlc = associationState.getClient().getInterfaceType().equals(InterfaceType.HDLC);
            ConnectorTask connectorTask = channel.attr(NettyConfigHelper.connectTaskRegular).get();
            this.sendPacket2Channel(channel, businessPacket);
            dlmsPacketCodecService.configRetryBO(isHdlc, channel, businessPacket, connectorTask, associationState.getDeviceAddress());
            //PLC表 AA建立成功后 更新上线状态
            if (connectorTask.getMeterId() != null && !terminal.getDeviceId().equals(connectorTask.getMeterId())) {
                deviceStatusService.refreshDeviceStatus(terminal, connectorTask.getMeterId(), ComEntity.CommunicationStatus.EXECUTING);
            }
        }
        if (aaAccepted == -1) {
            dlmsPacketCodecService.buildAAFailedCallBack(terminal, channel);
        }
    }

    public PacketType getPacketType(byte[] byteReceive, ComEntity terminal) {
        //ComEntity Channel Exists， handle packets
        ComConfigDMO comConfigDMO;
        if (terminal.getComId() == NettyConstant.ANONYMOUS_DEFAULT_COM_ID) {
            comConfigDMO = null;
        } else {
            comConfigDMO = assetManagementRemoteService.getComConfigById(terminal.getComId());
        }
        PacketType pt = null;
        try {
            pt = dlmsPacketCodecService.getPacketType(byteReceive, terminal, comConfigDMO);
        } catch (Exception e) {
            log.info("transfer failed size={}, packet={}", getFrameSize(byteReceive), ByteUtils.byteToHexString(byteReceive));
        }
        return pt;
    }

    private int getFrameSize(byte[] byteReceive) {
//        GXDLMSClient client = taskContext != null && taskContext.getClient() != null ? taskContext.getClient() : shareClient;
        GXByteBuffer buffer = new GXByteBuffer();
        buffer.set(byteReceive);
        return shareClient.getFrameSize(buffer);
    }


    @Override
    @DeviceSignInLimiter
    public byte[] deviceChannelEstablish(Channel channel, byte[] result) {
        //Check self register packet
        Tuple logicalAddr = dlmsPacketCodecService.getLogicalAddrFromLoggingFrame(result);
        if (logicalAddr == null || StringUtils.isEmpty(logicalAddr.get(0))) {
            try {
                channel.attr(NettyConfigHelper.closeBySystem).set(true);
                log.info("FailedGetDeviceAddress,disconnecting..");
                channel.disconnect().sync();
            } catch (InterruptedException e) {
            }
            return null;
        }
        selfService.doDeviceRegisterToSystem(channel, result, logicalAddr.get(0));
        return logicalAddr.get(1) == null ? new byte[0] : logicalAddr.get(1);
    }

    @Override
    public void doDeviceRegisterToSystem(Channel channel, byte[] result, String logicAddr) {
        ComEntity terminal = assetManagementRemoteService.getTerminalByLogicAddr(logicAddr);
        int localPort = NetUtil.getLocalPort(channel);
        String netAddress = NetUtil.getIpFromChannel(channel);
        Long deviceType = DeviceType.DCU.getId();
        if (localPort == Convert.toInt(p2pPort)) {
            deviceType = DeviceType.METER.getId();
        }

        if (terminal == null || terminal.getComId() == null) {
            // 记录匿名设备登录信息
            log.info("device in logic addr {} try to register. But asset information is not in UDIS", logicAddr);
            terminal = new ComEntity();
            terminal.setModuleSn(logicAddr);
            terminal.setIpAddress(netAddress);
            terminal.setComId(NettyConstant.ANONYMOUS_DEFAULT_COM_ID);

            anonyDeviceService.loginAnonyDevice(logicAddr, deviceType, netAddress);
            terminal.setComDeviceType(deviceType);
            sendAnonymousDeviceLoginDataMessageToMdm(deviceType, logicAddr, netAddress);
        } else {
            long now = SystemClock.now() / 1000;
            log.debug("tcp establish: comid={} ip={} terminal={}", terminal.getComId(), channel, terminal.getModuleSn());
            terminalChannelService.addReceiveRecord(terminal);
            deviceKeyManager.loadDefaultComKey(terminal);
            assetService.loadAssetToCache(terminal);
            channel.attr(NettyConfigHelper.deviceLgcTime).set(SystemClock.now());
            channel.attr(NettyConfigHelper.comEntityAttrKey).set(terminal);
            redisUtils.set(RedisKeys.getComLastHeartbeat(terminal.getComId()), String.valueOf(now), DeviceConnectServiceImpl.HEART_BEAT_VALID_SECONDS);
            channel.attr(NettyConfigHelper.hbTime).set(SystemClock.now());
            tcpClientManager.addConnectInfo(terminal.getComId());
            Boolean ipChanged = false;
            Boolean modeChanged = false;
            if (StrUtil.isEmpty(terminal.getIpAddress()) || !(netAddress.split(":")[0]).equalsIgnoreCase(terminal.getIpAddress().split(":")[0])) {
                log.info("com ip change: comid={} ip={}->{}", terminal.getComId(), terminal.getIpAddress(), netAddress);
                ipChanged = true;
                terminal.setIpAddress(netAddress);
                sendChangeIpMessageToMdm(netAddress, terminal);
            } else {
                //todo: 检查MDM上次IP是否发送成功 额外增加任务重试
//                // if send change ip to mdm
//                LogMdmSync lastMdmSyncLog = assetManagementRemoteService.getLastMdmSyncLog(terminal.getDeviceId(), terminal.getComDeviceType(), LogMdmSync.SyncType.IP.getValue(), netAddress);
//                if (lastMdmSyncLog == null || lastMdmSyncLog.getSyncResult() == LogMdmSync.Result.FAILED.getCode()) {
//                    sendChangeIpMessageToMdm(netAddress, terminal);
//                }
            }

            if (result != null) {
                String wsHbPacket = PacketAddressUtil.getHbPacketByDrliPacket(result);
                if (StrUtil.isNotEmpty(wsHbPacket)) {
                    // 登陆帧设置进channel里面存起来用于后续混合报文的CosemByteDecoder的解析
                    NettyConfigHelper.getChannelByAddress(netAddress).attr(NettyConfigHelper.wsDHBKey).set(wsHbPacket);
                }
            }
//            else {
//                // if send change ip to mdm
//                LogMdmSync lastMdmSyncLog = assetManagementRemoteService.getLastMdmSyncLog(terminal.getDeviceId(), terminal.getComDeviceType(), LogMdmSync.SyncType.IP.getValue(), netAddress);
//                if(lastMdmSyncLog == null || lastMdmSyncLog.getSyncResult() == LogMdmSync.Result.FAILED.getCode()) {
//                    sendChangeIpMessageToMdm(netAddress, terminal);
//                }
//            }

            if (terminal.getComMode().intValue() == ComEntity.CommunicationMode.CLIENT.getCode()) {
                log.info("com mode change: comid={},mode={}->{}", ComEntity.CommunicationMode.CLIENT.getCode(), ComEntity.CommunicationMode.SERVER.getCode());
                terminal.setComMode(ComEntity.CommunicationMode.SERVER.getCode());
                modeChanged = true;
            }

            if (ipChanged || modeChanged) {
                assetManagementRemoteService.updateComIpAddress(terminal.getComId(), netAddress, ComEntity.CommunicationMode.SERVER.getCode());
            }

//            comEntityStatusManager.comEntitySignIn(terminal, now, ipChanged, modeChanged);
        }
        this.unlockComEntity(channel);
        terminalChannelService.addTerminalNetAddress(terminal, channel);
        //把terminal信息放入connectionInfo
        connectionInfo.set(new ConnectionInfo().setModuleSn(terminal.getModuleSn()).setComId(terminal.getComId()));
    }

    private void sendChangeIpMessageToMdm(String netAddress, ComEntity terminal) {
        DeviceIpChangeEventDto deviceIpChangeEventDto = new DeviceIpChangeEventDto();
        deviceIpChangeEventDto.setIpAddress(netAddress);
        deviceIpChangeEventDto.setComId(terminal.getComId());
        deviceIpChangeEventDto.setTv(SystemClock.now() / 1000);
        kafkaTemplate.send(TaskKafkaTopic.CLIENT_DEVICE_IP_CHANGED, Convert.toStr(terminal.getComId()), JSONUtil.toJsonStr(deviceIpChangeEventDto).getBytes());
    }

    @Override
    public byte[] clientAccept(Channel channel, byte[] result) {
        String netAddress = NetUtil.getIpFromChannel(channel);
        UnwrappedPacket unwrappedPacket = dlmsPacketCodecService.unwrapPrefixPacket(result);
        SyncFuture future = terminalChannelService.getFutureSync(netAddress, unwrappedPacket.getDeviceAddress());
        if (future != null) {
            future.setResponse(result);
            return null;
        }
        ComEntity terminal = terminalChannelService.getTerminalByNetAddress(netAddress);
        if (terminal == null) {
            log.error("channel {} don't have terminal connection", netAddress);
            return null;
        }
        TaskContext context = terminalChannelService.getChannelLastSendContext(netAddress);
        //非同步请求的，处理返回报文
        return dlmsPacketCodecService.handleDeviceServicePacket(channel, terminal, netAddress, result, context).getResponse();
    }

    @Override
    public void sendPacket2Channel(Channel channel, byte[] packet) {
        MDC.put(CHANNEL, NetUtil.getIpFromChannel(channel));
        channel.writeAndFlush(packet);
    }

    @Override
    public void sendPacketDirect(ComEntity comEntity, byte[] hexStringToByte) {
        Channel channel = NettyConfigHelper.getChannelByComId(comEntity.getComId());
        sendPacket2Channel(channel, hexStringToByte);
    }

    @Override
    public void sendDeviceResultMessage(ConnectTaskResultDto connectTaskResultDto, ComEntity comEntity) {
        if (connectTaskResultDto.getTicketNumber() != null) {
            taskTicketManager.recycleTicket(connectTaskResultDto.getTicketNumber());
        }

//        Channel channelByComId = NettyConfigHelper.getChannelByComId(comEntity.getComId());
//        if (channelByComId != null) {
//            channelByComId.attr(NettyConfigHelper.connectTaskRegular).getAndSet(null);
//        }

        LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(comEntity), this.getClass()).info("AtomicTaskEnd. Status:{}", connectTaskResultDto.getExecuteCode());
        if (!LOCK_COM_ENTITY_FAILED.equalsIgnoreCase(connectTaskResultDto.getMessage())) {
            //无论TaskNo，获取到latch之后，countDown
            CountDownLatch comEntityResourceLatch = getComEntityResourceLatch(comEntity.getComId(), null);
            if (comEntityResourceLatch != null) {
                comEntityResourceLatch.countDown();
            } else {
                log.error("comIdExecuteTaskWithoutLatch:ComId:{}", comEntity.getComId());
            }
        }
        //客户端移除AtomicTaskNo锁
        if (ComEntity.CommunicationMode.CLIENT.getCode() == Optional.ofNullable(Optional.ofNullable(comEntity).orElse(new ComEntity()).getComMode()).orElse(-1)) {
            taskService.removeAtomicTaskKey(connectTaskResultDto.getAtomicTaskNo());
        }
//
        if ((connectTaskResultDto.getFlag() & 0x08) > 0) {
            //高优先级别使用REST回调
            try {
                this.sendDeviceResultRequest(connectTaskResultDto);
            } catch (Exception e) {
                log.debug("exceptionInCallBack", e);
                LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(comEntity), this.getClass()).info("SendToUTEfailed. at-no:{}", connectTaskResultDto.getAtomicTaskNo());
                this.sendDeviceResultMessage(connectTaskResultDto.getExecuteCode(), connectTaskResultDto.getAtomicTaskNo(), connectTaskResultDto.getContent(), connectTaskResultDto.getMessage(), connectTaskResultDto.getComId());
            }
        } else {
            this.sendDeviceResultMessage(connectTaskResultDto.getExecuteCode(), connectTaskResultDto.getAtomicTaskNo(), connectTaskResultDto.getContent(), connectTaskResultDto.getMessage(), connectTaskResultDto.getComId());
        }
    }

    private void sendDeviceResultRequest(ConnectTaskResultDto connectTaskResultDto) {
        ConnectorTaskResult.ExecuteCode code = connectTaskResultDto.getExecuteCode();
        ConnectorTaskResult connectorTaskResult = new ConnectorTaskResult();
        connectorTaskResult.setExecuteCode(code.getCode());
        connectorTaskResult.setResultData(connectTaskResultDto.getContent());
        connectorTaskResult.setAtomicTaskNo(connectTaskResultDto.getAtomicTaskNo());
        connectorTaskResult.setResultMessage(connectTaskResultDto.getMessage());
//        taskService.removeAtomicTaskKey(connectorTaskResult.getAtomicTaskNo());
        taskEngineService.responseTask(connectorTaskResult);
        log.debug("'task complete:' {} code={}, content={}, msg={} ", connectorTaskResult.getAtomicTaskNo(), code.getCode(), connectTaskResultDto.getContent(), connectTaskResultDto.getMessage());
    }

    private void sendDeviceResultMessage(ConnectorTaskResult.ExecuteCode code, String atomicTaskNo, String content, String message, Long comId) {
        ConnectorTaskResult connectorTaskResult = new ConnectorTaskResult();
        connectorTaskResult.setExecuteCode(code.getCode());
        connectorTaskResult.setResultData(content);
        connectorTaskResult.setAtomicTaskNo(atomicTaskNo);
        connectorTaskResult.setResultMessage(message);
        kafkaTemplate.send(TaskKafkaTopic.TASK_SEND_TO_UTE, Convert.toStr(comId), ProtoBufferUtil.toProtoBuffer(connectorTaskResult, ConnectorTaskResultProto.ConnectorTaskResult.class).toByteArray());
//        taskService.removeAtomicTaskKey(atomicTaskNo);
        log.debug("'task complete:' {} code={}, content={}, msg={} ", atomicTaskNo, code.getCode(), content, message);
    }

    private void sendAnonymousDeviceLoginDataMessageToMdm(Long deviceType, String logicAddr, String netAddress) {
        MdmAnonyTerminalDto anonyTerminalDto = new MdmAnonyTerminalDto();
        anonyTerminalDto.setTv(SystemClock.now() / 1000);
        anonyTerminalDto.setLogidAddr(logicAddr);
        anonyTerminalDto.setDataType(0);
        anonyTerminalDto.setTerminalStatus(MdmAnonyTerminalDto.TerminalStatus.Online.getCode());
        if (deviceType.equals(DeviceType.DCU.getId())) {
            anonyTerminalDto.setTerminalType(MdmAnonyTerminalDto.TerminalType.DCU.getCode());
        } else {
            anonyTerminalDto.setTerminalType(MdmAnonyTerminalDto.TerminalType.P2P_METER.getCode());
        }
        anonyTerminalDto.setNetAddr(netAddress);
        anonyTerminalDto.setDataSource(Convert.toStr(MdmSourceEnum.ScheduleRead.getCode()));
        anonyTerminalDto.setValidity(0);
        anonyTerminalDto.setTvUpdate(SystemClock.now() / 1000);
        kafkaTemplate.send(TaskKafkaTopic.ANONYMOUS_DEVICE_STATUS_TRANSFER, JSONUtil.toJsonStr(anonyTerminalDto).getBytes());
    }

    public static UnwrappedPacket unwrapPrefixPacket(byte[] packet) {
        GXDLMSTranslator translator = new GXDLMSTranslator(TranslatorOutputType.SIMPLE_XML);
        UnwrappedPacket unwrappedPacket = new UnwrappedPacket();
        unwrappedPacket.setPacket(packet);

        try {
            byte[] pdu = translator.getPdu(packet);
            if (pdu == null || pdu.length == 0) {
                return unwrappedPacket;
            }
            if (pdu[0] == (byte) 0xe7) {
                int wrapIndex = Bytes.indexOf(packet, pdu);
                byte[] prefixPduLen = new byte[2];
                System.arraycopy(packet, wrapIndex - 2, prefixPduLen, 0, 2);

                int totalPduLength = ByteNumberUtils.byte2ToShort(prefixPduLen);

                //获取第3个字节的设备地址长度
                byte prefixLen = pdu[2];
                byte[] deviceAddress = new byte[prefixLen];
                System.arraycopy(pdu, 3, deviceAddress, 0, prefixLen);
                String meterNo = new String(ByteUtils.getChars(deviceAddress));
                unwrappedPacket.setDeviceAddress(meterNo);
                byte[] wrapper = new byte[wrapIndex];
                System.arraycopy(packet, 0, wrapper, 0, wrapIndex);

                int totalPrefixLen = 3 + prefixLen;
                int devicePduLength = totalPduLength - totalPrefixLen; // e7 00 0x {device info} {device pdu}
                byte[] devicePdu = new byte[devicePduLength];
                System.arraycopy(packet, wrapIndex + totalPrefixLen, devicePdu, 0, devicePduLength);

                byte[] wrapLength = ByteNumberUtils.shortToByte2((short) devicePduLength);
                System.arraycopy(wrapLength, 0, wrapper, wrapper.length - 2, 2);
                unwrappedPacket.setPacket(Bytes.concat(wrapper, devicePdu));
            }
        } catch (Exception e) {
            log.error("unwrap fail , packet {}", ByteUtils.byteToHexString(packet));
            return null;
        }

        return unwrappedPacket;
    }

}
